package com.ydlclass.proxy.handler;

import com.ydlclass.NettyBootstrapInitializer;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.annotation.TryTimes;
import com.ydlclass.compress.CompressorFactory;
import com.ydlclass.discovery.Registry;
import com.ydlclass.enumeration.RequestType;
import com.ydlclass.exceptions.DiscoveryException;
import com.ydlclass.exceptions.NetworkException;
import com.ydlclass.protection.CircuitBreaker;
import com.ydlclass.serialize.SerializerFactory;
import com.ydlclass.transport.message.RequestPayload;
import com.ydlclass.transport.message.YrpcRequest;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Date;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程，都封装在invoke方法中
 * 1、发现可用服务  2、建立连接  3、发送请求  4、得到结果
 */

@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {

    //此处需要一个注册中心 和 一个接口
    private final Registry registry;
    private final Class<?> interfaceRef;

    private String group;
    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceRef,String group) {
        this.registry = registry;
        this.interfaceRef = interfaceRef;
        this.group = group;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 1.从接口中判断是否需要重试
        TryTimes tryTimesAnnotation = method.getAnnotation(TryTimes.class);
        // 默认值0，代表不重试
        int tryTimes = 0;
        int intervalTime = 0;
        if(tryTimesAnnotation != null){
            tryTimes = tryTimesAnnotation.tryTimes();
            intervalTime = tryTimesAnnotation.intervalTime();
        }

        while(true) {
            // 什么时候重试  1、异常 2、响应出问题  code == 500

            /**
             * -------------------1、封装报文-------------------
             */
            RequestPayload requestPayload = RequestPayload.builder()
                    .interfaceName(interfaceRef.getName())
                    .methodName(method.getName())
                    .parametersType(method.getParameterTypes())
                    .parametersValue(args)
                    .returnType(method.getReturnType())
                    .build();
            // 创建一个请求
            YrpcRequest yrpcRequest = YrpcRequest.builder()
                    .requestId(YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                    .compressType(CompressorFactory.getCompressor(YrpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                    .requestType(RequestType.REQUEST_TYPE.getId())
                    .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                    .timeStamp(System.currentTimeMillis())
                    .requestPayload(requestPayload)
                    .build();

            // 将请求存入本地线程需要在合适的时候，调用remove方法
            YrpcBootstrap.REQUEST_THREAD_LOCAL.set(yrpcRequest);


            // 2、发现服务，从注册中心拉取服务列表，并通过客户端负载均衡寻找一个可用的服务
            // 传入服务的名字，返回ip+端口
            InetSocketAddress address = YrpcBootstrap.getInstance().getConfiguration().getLoadBalancer().selectServiceAddress(interfaceRef.getName(),group);
            if (log.isDebugEnabled()) {
                log.debug("服务调用方，发现了服务{}的可用主机{}", interfaceRef.getName(), address);
            }

            //获取当前地址所对应的断路器
            Map<SocketAddress, CircuitBreaker> everyIpCircuitBreaker = YrpcBootstrap.getInstance().
                    getConfiguration().getEveryIpCircuitBreaker();
            CircuitBreaker circuitBreaker = everyIpCircuitBreaker.get(address);
            if(circuitBreaker == null){
                circuitBreaker = new CircuitBreaker(10,0.5F);
                everyIpCircuitBreaker.put(address,circuitBreaker);
            }
            try {
                //3.如果断路器是打开的
                if(yrpcRequest.getRequestType() != RequestType.HEART_TYPE.getId() && circuitBreaker.isBreak()){
                    // 定期关闭
                    Timer timer = new Timer();
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            YrpcBootstrap.getInstance().
                                    getConfiguration().getEveryIpCircuitBreaker().get(address).reset();
                        }
                    },5000);

                    throw new RuntimeException("当前断路器已经开启，无法发送请求.");
                }


                // 使用netty连接服务器，发送调用的 服务器的名字+方法名字+参数列表，得到最后结果
                //定义线程池 EventLoopGroup
                // q:整个连接过程放在这里，意味着每次调用都会产生一个新的netty连接。 那如何缓存我们的连接

                // 解决方案：缓存channel，尝试从缓存中获取channel，如果未获取，则创建新的连接，进行缓存。
                // 4.尝试获取一个可用通道
                Channel channel = getAvailableChannel(address);
                if (log.isDebugEnabled()) {
                    log.debug("获取了和{}建立的连接通道,准备发送数据", address);
                }

                /**
                 * -------------------同步策略-------------------
                 */
//                ChannelFuture channelFuture = channel.writeAndFlush(new Object()).await();
//                //学习channelFuture的简单api get方法是阻塞的，getNow 获取当前结果，如果处理未完成，则返回null
//                if(channelFuture.isDone()){
//                    Object object = channelFuture.getNow();
//                } else if (!channelFuture.isSuccess()) {
//                    //需要捕获异常,可以捕获异步任务中的异常
//                    Throwable cause = channelFuture.cause();
//                    throw new RuntimeException(cause);
//                }

                /**
                 * -------------------异步策略-------------------
                 */

                //5.写出报文
                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                //将completableFuture暴露出去
                YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(), completableFuture);


                // 这里 yrpcRequest 写出一个请求，这个请求的实例就会进入pipeline执行出站的一系列操作
                // 第一个出站程序一定是将 yrpcRequest ---> 二进制报文
                channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {

                    //当前的promise返回的结果是writeAndFlush的返回结果
                    //一旦数据被写出去，promise就结束了
                    //我们其实想要的是服务端给我们的返回值，所以不能这样
                    //需要将completableFuture 挂起暴露，并且在得到服务提供方响应时，调用complete方法

//                    if(promise.isDone()){
//                        completableFuture.complete(promise.getNow());
//                    }
                    if (!promise.isSuccess()) {
                        completableFuture.completeExceptionally(promise.cause());
                    }
                });

                // 6.清理threadLocal
                YrpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                // 如果没有地方处理这个 CompletableFuture， 这里会阻塞等待 complete 方法的执行。
                // q:我们需要在哪里调用complete方法得到结果，很明显 在pipeline 中最终的handler的处理结果
                // 7.获得响应的结果
                Object result = completableFuture.get(50, TimeUnit.SECONDS);
                // 记录成功的请求
                circuitBreaker.recordRequest();
                return result;
            } catch (Exception e) {
                // 如果捕获到了异常，次数-1，等待固定时间后重试(有可能会出现重试风暴)
                tryTimes--;
                // 记录错误的次数
                circuitBreaker.recordErrorRequest();
                try{
                    Thread.sleep(intervalTime);
                }catch (InterruptedException ex){

                }
                if(tryTimes < 0){
                    log.error("对方法{}进行远程调用时，重试{}次，依然不可调用"
                            ,method.getName(),tryTimes,e);
                    break;
                }
                log.error("在进行第{}次重试时发生异常，",3-tryTimes,e);
            }
        }
        throw new RuntimeException("执行远程方法"+method.getName()+"调用失败。");
    }


    /**
     * 根据地址获取一个可用的通道
     * @param address
     * @return
     */
    private Channel getAvailableChannel(InetSocketAddress address) {
        //1.尝试从缓存中获取
        Channel channel = YrpcBootstrap.CHANNEL_CACHE.get(address);

        //2.拿不到就去建立连接
        if(channel == null){
            //await 方法会阻塞，会等待连接成功再返回，netty还提供了异步处理的逻辑
            //sync和await都是阻塞当前线程，获取返回值(连接的过程是异步的，发送数据的过程也是异步的)
            //如果发生了异常，sync会主动在主线程抛出异常，await不会，异常在子线程中处理需要使用future
//                    channel = NettyBootstrapInitializer.getBootstrap().connect(address).await().channel();

            // 使用addListener执行的异步操作
            CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener(
                    (ChannelFutureListener) promise -> {
                        if(promise.isDone()){
                            // 异步的，已经完成
                            if(log.isDebugEnabled()){
                                log.debug("已经和{}成功建立了连接",address);
                            }
                            channelFuture.complete(promise.channel());
                        }else if(!promise.isSuccess()){
                            channelFuture.completeExceptionally(promise.cause());
                        }
                    }
            );

            //阻塞获取channel
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("获取通道时，发生异常。",e);
                throw new DiscoveryException(e);
            }

            //缓存channel
            YrpcBootstrap.CHANNEL_CACHE.put(address,channel);
        }

        if(channel == null){
            log.error("获取或建立与{}的通道时，发生了异常.",address);
            throw new NetworkException("获取通道时，发生了异常.");
        }
        return channel;
    }
}
